package com.audaque.springboot.foshanupload.web.rocketmqdemo.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.*;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

@Slf4j
@RestController
@RequestMapping("clientConsumer")
public class ClientConsumerController {


    @Autowired
    private ClientConfiguration clientConfiguration;

    @GetMapping("push")
    public void push(
            @RequestParam("topic") String topic, // 消息发送的目标Topic名称，需要提前创建。
            @RequestParam("consumerGroup") String consumerGroup,// 为消费者指定所属的消费者分组，Group需要提前创建
            @RequestParam(value = "messageTag",required = false) String messageTag// 订阅消息的过滤规则，表示订阅所有Tag的消息
    ) throws ClientException, IOException {
        // Credential provider is optional for client configuration.
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息
        if (StringUtils.isBlank(messageTag)) {
            messageTag = "*";
        }
        FilterExpression filterExpression = new FilterExpression(messageTag, FilterExpressionType.TAG);
        // 初始化 PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系
        PushConsumer consumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // 设置消费者分组
                .setConsumerGroup(consumerGroup)
                // 设置预绑定的订阅关系
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // 设置消费监听器
                .setMessageListener(messageView -> {
                    // 处理消息并返回消费结果
                    log.debug("consume message successfully, messageId={}", messageView.getMessageId());
                    // 消息内容处理
                    ByteBuffer body = messageView.getBody();
                    String message = StandardCharsets.UTF_8.decode(body).toString();
                    body.flip();
                    log.debug("message body={}", message);
                    return ConsumeResult.SUCCESS;
                }).build();
        // 如果不需要再使用 consumer，可关闭该实例。
        //consumer.close();
    }


    @GetMapping("simple")
    public void simple(
            @RequestParam("topic") String topic, // 消息发送的目标Topic名称，需要提前创建。
            @RequestParam("consumerGroup") String consumerGroup,// 为消费者指定所属的消费者分组，Group需要提前创建
            @RequestParam(value = "messageTag",required = false) String messageTag,// 订阅消息的过滤规则，表示订阅所有Tag的消息
            @RequestParam("awaitSecond") Integer awaitSecond,//拉取周期
            @RequestParam("maxMessageNum") Integer maxMessageNum,//拉取条数
            @RequestParam("invisibleSecond") Integer invisibleSecond  //消费超时
    ) throws ClientException, IOException {
        // 订阅消息的过滤规则，表示订阅所有Tag的消息
        if (StringUtils.isBlank(messageTag)) {
            messageTag = "*";
        }
        FilterExpression filterExpression = new FilterExpression(messageTag, FilterExpressionType.TAG);
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setAwaitDuration(Duration.ofSeconds(awaitSecond))
                .build();
        List<MessageView> messageViewList = null;
        try {
            messageViewList = consumer.receive(maxMessageNum, Duration.ofSeconds(invisibleSecond));
            log.debug("Received {} message(s)", messageViewList.size());
            messageViewList.forEach(messageView -> {
                // 消息内容处理
                ByteBuffer body = messageView.getBody();
                String message = StandardCharsets.UTF_8.decode(body).toString();
                body.flip();
                log.debug("message body={}", message);

                //消费处理完成后，需要主动调用ACK提交消费结果
                try {
                    consumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求
            e.printStackTrace();
        }
        // 如果不需要再使用 consumer，可关闭该实例。
        //consumer.close();
    }


}


